package com.hyt.it.ogt.kq.service.bm.command;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.hyt.common.apiInfor.ApiResponse;
import com.hyt.common.utils.exception.BusinessException;
import com.hyt.core.constant.CommonConstant;
import com.hyt.core.enums.NumberEnum;
import com.hyt.it.ogt.kq.service.bm.feign.tps.TpsClient;
import com.hyt.it.ogt.kq.service.bm.feign.tps.model.WxOffiaccountTemplateMsg;
import com.hyt.it.ogt.kq.service.bm.model.dto.OpenIdReflectStudentDTO;
import com.hyt.it.ogt.kq.service.model.vo.wechat.PushExamNotifyDTO;
import com.hyt.it.ogt.kq.service.bm.model.entity.Project;
import com.hyt.it.ogt.kq.service.bm.service.IProjectService;
import com.hyt.it.ogt.kq.service.bm.service.IStudentProjectService;
import com.hyt.it.ogt.kq.common.bm.utils.DateUtil;
import com.hyt.loginfo.model.UserExtraDTO;
import com.hyt.loginfo.service.IUaUserExtService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @program: it-ogt-bm
 * @description: 微信小程序通知推送
 * @author: Yang HuanKun
 * @create: 2022-01-12 15:29
 **/
@Slf4j
@Component
public class WeChatMiniNotifyPushCommand extends DestroyBean {

    /**
     * 微信小程序通知推送队列redis的key
     */
    public static final String WECHAT_MINI_NOTIFY_PUSH_QUEUE_KEY = "wechat:mini:push";

    /**
     * 线程池最大线程数
     */
    private static final int MAX_POOL_SIZE = NumberEnum.TEN.getK();

    /**
     * 推送消息线程池，重写拒绝策略让消息使命必达
     */
    private static final ExecutorService PUSH_MESSAGE_EXECUTOR_SERVICE = new ThreadPoolExecutor(
            CommonConstant.ONE,
            MAX_POOL_SIZE,
            0L,
            TimeUnit.MILLISECONDS,
            new SynchronousQueue<>(true),
            new ThreadFactoryBuilder()
                    .setNamePrefix("push-wechat-message-pool-thread-")
                    .build(),
            new RejectedExecutionHandler() {
                // 注意，此处把线程池中的阻塞队列拿出来，拒绝策略重新put Runnable
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                	BlockingQueue<Runnable> blockingQueue = executor.getQueue();
                    if (ObjectUtil.isNotNull(blockingQueue)) {
                        try {
							blockingQueue.put(r);
						} catch (InterruptedException e) {
							log.error("##  threadPoolExecutor init error", e);
							Thread.currentThread().interrupt();
						}
                    }
                }
            });

    @Resource
    RedisTemplate<String, PushExamNotifyDTO> redisTemplate;
    @Resource
    TpsClient tpsClient;
    @Resource
    IStudentProjectService studentProjectService;
    @Resource
    IUaUserExtService uaUserExtService;
    @Resource
    IProjectService projectService;

    @Override
    public void run(String... args) throws Exception {
        PUSH_MESSAGE_EXECUTOR_SERVICE.execute(() -> {
            // 这里使用线程池控制while循环状态的原因是防止已经调用了线程池的shutdown()，由于任务的原因还在继续拉取推送
            while (!PUSH_MESSAGE_EXECUTOR_SERVICE.isShutdown()) {
                PUSH_MESSAGE_EXECUTOR_SERVICE.execute(this::pushWeChatNotify);
            }
        });
        log.info("# weChat mini notify push run ...");
    }

    /**
     * 推送微信通知
     */
    private synchronized void pushWeChatNotify() {
        List<PushExamNotifyDTO> pushExamNotifyDTOS = null;
        try {
            // 获取推送通知数据集合
            pushExamNotifyDTOS = this.leftPopPushExamNotify();
            // 如果获取到的推送列表为空则阻塞跳出等待重新获取
            if (CollUtil.isEmpty(pushExamNotifyDTOS)) {
                wait(200);
                return;
            }

            // 获取考生对应openID的映射
            List<OpenIdReflectStudentDTO> openIdReflectStudentDTO = this.getOpenIdReflectStudentDTO(pushExamNotifyDTOS);

            // 转化用户的openID对应的map格式为 Map<StudentId, OpenId>
            Map<String, String> studentReflectOpenIdMap = openIdReflectStudentDTO.stream()
                    .parallel()
                    .filter(data -> StrUtil.isNotBlank(data.getStudentId()))
                    .filter(data -> StrUtil.isNotBlank(data.getOpenId()))
                    .collect(Collectors.toMap(OpenIdReflectStudentDTO::getStudentId, OpenIdReflectStudentDTO::getOpenId));

            // 转化用户的officeID对应的map格式为 Map<StudentId, officeId>
            Map<String, String> studentReflectOfficeIdMap = openIdReflectStudentDTO.stream()
                    .parallel()
                    .filter(data -> StrUtil.isNotBlank(data.getStudentId()))
                    .filter(data -> StrUtil.isNotBlank(data.getOfficeId()))
                    .collect(Collectors.toMap(OpenIdReflectStudentDTO::getStudentId, OpenIdReflectStudentDTO::getOfficeId));

            // 这里组装推到tps服务的实体参数
            List<WxOffiaccountTemplateMsg> wxOffiaccountTemplateMsgs = pushExamNotifyDTOS.stream()
                    .parallel()
                    .map(data -> {
                        // 从map中获取考生对应的openID
                        String toUser = studentReflectOpenIdMap.get(data.getId());
                        // 从map中获取考生对应的officeID
                        String officeId = studentReflectOfficeIdMap.get(data.getId());

                        return WxOffiaccountTemplateMsg.builder()
                                .toUser(toUser)
                                .officeId(officeId)
                                .data(this.getSubTemplateItem(data))
                                .build();
                    }).collect(Collectors.toList());

            tpsClient.pushWeChatOffiAccountTemplateMessage(wxOffiaccountTemplateMsgs);
        } catch (QueryTimeoutException e) {
            // 抓这个异常不打印的原因用于redis list pop 阻塞设定阻塞超时会抛出查询超时异常
        } catch (BusinessException e) {
            log.error("# 通知推送数据缺失异常信息：", e);
        } catch (InterruptedException e) {
            // 这里做推送到tps服务失败情况下错误补偿机制
            this.redressException(pushExamNotifyDTOS);
            log.error("# 微信小程序通知推送异常：", e);
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 获取考生对应openID的映射
     *
     * @param pushExamNotifyDTOS
     * @return
     */
    private List<OpenIdReflectStudentDTO> getOpenIdReflectStudentDTO(List<PushExamNotifyDTO> pushExamNotifyDTOS) {
        // 获取考生ID集合
        List<String> studentIds = pushExamNotifyDTOS.stream()
                .map(PushExamNotifyDTO::getId)
                .distinct()
                .collect(Collectors.toList());
        // Map<UserId, StudentId> 获取用户的关联信息通过考生ID
        Map<String, String> userIdsMap = Optional.ofNullable(studentProjectService.getRegIdById(studentIds))
                .filter(CollUtil::isNotEmpty)
                .orElseThrow(() -> new BusinessException("未查询到存在考生信息"));

        // Map<StudentId, OfficeId> 通过考试ID获取对应的机构ID
        Map<String, String> officeIdReflectStudent = this.getOfficeIdReflectStudent(studentIds);

        // Map<UserId, OpenId> 获取用户ID对应的openID 
        Set<String> userIds = userIdsMap.keySet();
        Map<String, UserExtraDTO> openIdsMap = Optional.ofNullable(uaUserExtService.getWeChatOffiAccountWeChatOpenIds(userIds))
                .map(ApiResponse::getData)
                .filter(MapUtil::isNotEmpty)
                .orElseThrow(() -> new BusinessException("远程ua服务获取用户微信信息拉取为空"));

        // 获取openID映射实体
        List<OpenIdReflectStudentDTO> openIdReflectStudentDTOS = openIdsMap.entrySet()
                .stream()
                .map(data -> {
                    String userId = data.getKey();
                    // 从Map<UserId, StudentId>中通过用户ID获取学生ID
                    String studentId = userIdsMap.get(data.getKey());
                    // 从Map<StudentId, OfficeId>中通过学生ID获取机构ID
                    String officeId = officeIdReflectStudent.get(studentId);

                    OpenIdReflectStudentDTO.OpenIdReflectStudentDTOBuilder builder = OpenIdReflectStudentDTO.builder();
                    Optional.ofNullable(data.getValue())
                            .ifPresent(userExtraDTO -> {
                                builder.studentId(studentId)
                                        .userId(userId)
                                        .openId(userExtraDTO.getWeixinOpenId())
                                        .officeId(officeId);
                            });
                    return builder.build();
                }).collect(Collectors.toList());

        if (CollUtil.isEmpty(openIdReflectStudentDTOS)) {
            throw new BusinessException("考生微信数据信息拉取为空");
        }

        return openIdReflectStudentDTOS;
    }

    /**
     * 通过考试ID获取对应的机构ID
     *
     * @param studentIds 考生ID集合
     * @return Map<StudentId, OfficeId>
     */
    private Map<String, String> getOfficeIdReflectStudent(List<String> studentIds) {
        // Map<ProjectId, StudentId> 获取项目ID和考生ID绑定的Map
        Map<String, String> studentIdReflectProjectId = Optional.ofNullable(studentProjectService.getProjectIdById(studentIds))
                .filter(CollUtil::isNotEmpty)
                .orElseThrow(() -> new BusinessException("未查询到存在考生所对应的目标项目"));

        // Map<StudentId, OfficeId> 获取机构ID
        Map<String, String> studentIdReflectOfficeId = Optional.ofNullable(projectService.listByIds(studentIdReflectProjectId.keySet()))
                .orElseThrow(() -> new BusinessException("未查询到项目数据"))
                .stream()
                .filter(data -> StrUtil.isNotBlank(data.getId()))
                .filter(data -> StrUtil.isNotBlank(data.getOfficeId()))
                .collect(Collectors.toMap(data -> {
                    // 这里是获取学生ID
                    return studentIdReflectProjectId.get(data.getId());
                }, Project::getOfficeId, (oldValue, newValue) -> newValue));
        return studentIdReflectOfficeId;
    }

    /**
     * 获取redis列表中的推送数据
     * 一次获取${wechat.appMini.pushExamUserNotifyNum}条
     *
     * @return
     */
    private List<PushExamNotifyDTO> leftPopPushExamNotify() {
        // 获取集合基础大小负载因子，防止超出扩容
        int defaultPushExamNotifyCapacity = this.getDefaultPushExamNotifyCapacity();
        List<PushExamNotifyDTO> pushExamNotifyDTOS = Lists.newArrayListWithCapacity(defaultPushExamNotifyCapacity);
        // 循环拿取100个推送，如果有获取空就跳出
        for (int i = 0; i < 100; i++) {
            PushExamNotifyDTO pushExamNotifyDTO = redisTemplate.boundListOps(WECHAT_MINI_NOTIFY_PUSH_QUEUE_KEY)
                    .leftPop();
            if (ObjectUtil.isNull(pushExamNotifyDTO)) {
                break;
            }
            pushExamNotifyDTOS.add(pushExamNotifyDTO);
        }
        return pushExamNotifyDTOS;
    }

    /**
     * 获取默认的推送通知集合大小
     *
     * @return
     */
    private int getDefaultPushExamNotifyCapacity() {
        return 100 + (100 >> 1);
    }

    /**
     * 获取推送模板项
     *
     * @param pushExamNotifyDTO
     * @return
     */
    private WxOffiaccountTemplateMsg.SubTemplateItem getSubTemplateItem(PushExamNotifyDTO pushExamNotifyDTO) {
        String examDate = DateUtil.mergeDate(pushExamNotifyDTO.getStartTime(), pushExamNotifyDTO.getEndTime());
        WxOffiaccountTemplateMsg.SubTemplateItem subTemplateItem = new WxOffiaccountTemplateMsg.SubTemplateItem()
                .putItem("keyword1", pushExamNotifyDTO.getName())
                .putItem("keyword2", pushExamNotifyDTO.getSubjectName())
                .putItem("keyword3", examDate);
        if (StrUtil.isNotBlank(pushExamNotifyDTO.getAddress())) {
            subTemplateItem.putItem("remark", pushExamNotifyDTO.getAddress());
        }
        return subTemplateItem;
    }

    /**
     * 这里做推送到tps服务失败情况下错误补偿机制
     *
     * @param pushExamNotifyDTOS
     */
    private void redressException(List<PushExamNotifyDTO> pushExamNotifyDTOS) {
        try {
            Optional.ofNullable(pushExamNotifyDTOS)
                    .filter(CollUtil::isNotEmpty)
                    .map(data -> pushExamNotifyDTOS.toArray(new PushExamNotifyDTO[CommonConstant.ZERO]))
                    .ifPresent(data -> {
                        redisTemplate.boundListOps(WECHAT_MINI_NOTIFY_PUSH_QUEUE_KEY).rightPushAll(data);
                    });
        } catch (Exception e) {
            log.error("# WeChat Mini Notify Push Redress Exception", e);
        }
    }

    /**
     * 这里做推送到tps服务失败情况下错误补偿机制
     *
     * @param pushExamNotifyDTO
     */
    @SuppressWarnings("unused")
	private void redressException(PushExamNotifyDTO pushExamNotifyDTO) {
        try {
            log.info("# 进入补偿的推送数据：{}", pushExamNotifyDTO);
            Optional.ofNullable(pushExamNotifyDTO)
                    .ifPresent(data -> redisTemplate.boundListOps(WECHAT_MINI_NOTIFY_PUSH_QUEUE_KEY).rightPush(data));
        } catch (Exception e) {
            log.error("# WeChat Mini Notify Push Redress Exception", e);
        }
    }

    
    
    /**
     * 这里需要等待2秒钟
     *
     * @throws Exception
     */
    @Override
    public void destroy() {
        PUSH_MESSAGE_EXECUTOR_SERVICE.shutdown();
        try {
            // 这里进行等候一定时间给正在执行的任务执行完成，这里等待时间原则上应该以feign请求tps服务timeout为准
            if (!PUSH_MESSAGE_EXECUTOR_SERVICE.awaitTermination(CommonConstant.TWO, TimeUnit.SECONDS)) {
                PUSH_MESSAGE_EXECUTOR_SERVICE.shutdownNow();
            }
        } catch (InterruptedException ie) {
            PUSH_MESSAGE_EXECUTOR_SERVICE.shutdownNow();
            Thread.currentThread().interrupt();
        }
        log.info("# WeChat mini notify push SingleExecutor destroy ...");
    }
}

